package com.chao.elasticsearch.config;

import cn.hutool.core.collection.CollUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHeaders;
import org.apache.http.HttpHost;
import org.apache.http.HttpResponseInterceptor;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.entity.ContentType;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.springframework.boot.autoconfigure.elasticsearch.ElasticsearchProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Optional;

@Configuration
@EnableConfigurationProperties(ElasticsearchProperties.class)
public class ElasticsearchConfig {

    @Bean
    public ElasticsearchClient elasticsearchClient(ElasticsearchProperties elasticsearchProperties) {
        HttpHost[] httpHosts = toHttpHost(elasticsearchProperties.getUris());
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(
                AuthScope.ANY, new UsernamePasswordCredentials(elasticsearchProperties.getUsername(),elasticsearchProperties.getPassword()));

        RestClientBuilder builder = RestClient.builder(httpHosts);
        builder.setRequestConfigCallback(
                requestConfigBuilder -> requestConfigBuilder.setSocketTimeout((int) elasticsearchProperties.getSocketTimeout().toMillis())
                        .setConnectTimeout((int) elasticsearchProperties.getConnectionTimeout().toMillis()));
        builder.setHttpClientConfigCallback(httpAsyncClientBuilder -> httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider)
                .addInterceptorLast(
                        (HttpResponseInterceptor)
                                (response, context) ->
                                        response.addHeader("X-Elastic-Product", "Elasticsearch")));
        RestClient restClient = builder.build();
        ElasticsearchTransport transport = new RestClientTransport(restClient,new JacksonJsonpMapper());
        return new ElasticsearchClient(transport);
    }

    private HttpHost[] toHttpHost(List<String> urls) {
        if (CollUtil.isEmpty(urls)) {
            throw new RuntimeException("至少配置一个es的客户端连接!");
        }
        return Optional.of(urls)
                .map(urls1 -> urls1.stream()
                        .map(HttpHost::create)
                        .toArray(HttpHost[]::new))
                .orElseThrow(() -> new RuntimeException("创建es客户端失败！"));
    }
}
